package main

import (
	"fmt"
	"github.com/IBM/sarama"
	"log"
)

func main() {
	// producer 配置项
	config := sarama.NewConfig()
	// 1. 签收模式
	config.Producer.RequiredAcks = sarama.WaitForAll
	// 2. 发送负载策略
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	// 3. 交付成功消息
	config.Producer.Return.Successes = true

	// 连接kafka
	client, err := sarama.NewSyncProducer([]string{"8.140.22.220:9092"}, config)
	if err != nil {
		log.Fatalln(err)
		return
	}
	// 准备释放
	defer func() { client.Close() }()

	const topic = "topic01"
	for i := 0; i < 20; i++ {
		msg := &sarama.ProducerMessage{}
		msg.Topic = topic
		msg.Value = sarama.StringEncoder(fmt.Sprintf("new-msg11-%d", i))
		part, offset, err := client.SendMessage(msg)
		if err != nil {
			fmt.Println("send msg failed, err:", err)
			return
		}
		fmt.Printf("part:%v offset:%v\n", part, offset)
	}

}
